package com.cxl.kafka;

import cn.hutool.core.util.RandomUtil;
import cn.hutool.http.HttpRequest;
import com.alibaba.fastjson.JSON;
import com.cxl.entity.Vip;
import com.cxl.entity.Yuyue;
import com.cxl.service.IVipService;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;

/**
 * @author: 20971
 * @time: 2022/2/17 10:17
 * @description:
 */
@Component
public class KafkaConsumer {


    @Autowired
    private RedisTemplate redisTemplate;

    @Autowired
    private IVipService vipService;

    @KafkaListener(topics = {"test"})
    public void listener(ConsumerRecord<String, String> consumerRecord, Acknowledgment acknowledgment) {
        String key = consumerRecord.key();
        String value = consumerRecord.value();
        if (StringUtils.isNoneBlank(key) && StringUtils.isNoneBlank(value)) {
            Yuyue yuyue = JSON.parseObject(value, Yuyue.class);
            //防止重复消费
            if (!redisTemplate.opsForValue().setIfAbsent(value, null, 10, TimeUnit.HOURS)) {
                if (key.equals("yuyue")) {
                    String code = RandomUtil.randomNumbers(4);
                    Vip byId = vipService.getById(yuyue.getVid());
                    Map<String, Object> map = new HashMap<>();
                    map.put("number", byId.getPhone());
                    map.put("code",code);
                    //发送短信
                    String body = HttpRequest.post("http://192.168.182.130:8281/message/shortmessage")
                            .form(map)
                            .execute()
                            .body();
                    System.out.println(body);
                }
            }
            //防止消息丢失
            redisTemplate.delete(value);
        }
        //手动提交
        acknowledgment.acknowledge();
    }
}
